package com.gitee.cheungming.search.canal;

import cn.hutool.json.JSONUtil;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.gitee.cheungming.search.doc.GoodsDocument;
import com.gitee.cheungming.search.reposity.GoodsDocumentRepository;
import jakarta.annotation.PreDestroy;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.annotation.Bean;
import org.springframework.context.event.EventListener;
import org.springframework.core.task.TaskExecutor;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import java.math.BigDecimal;
import java.net.InetSocketAddress;
import java.util.List;
import java.util.concurrent.ThreadPoolExecutor;

@Component
@Slf4j
public class CanalHandler {

    @Resource
    private GoodsDocumentRepository goodsDocumentRepository;

    private volatile boolean running = false;
    private Thread canalThread;

    @EventListener(ApplicationReadyEvent.class)  // 改用这个注解
    public void startCanal() {
        running = true;
        canalThread = new Thread(this::runCanal, "canal-client-thread");
        canalThread.setDaemon(true);  // 设置为守护线程，不影响应用关闭
        canalThread.start();
        log.info("Canal 客户端启动完成");
    }

    @PreDestroy
    public void stopCanal() {
        running = false;
        if (canalThread != null) {
            canalThread.interrupt();
        }
        log.info("Canal 客户端已停止");
    }

    private void runCanal() {
        CanalConnector connector = CanalConnectors.newSingleConnector(
                new InetSocketAddress("192.168.50.2", 11111), "example", "", "");

        try {
            connector.connect();
            connector.subscribe("mall_goods.goods");
            connector.rollback();
            log.info("Canal 连接成功，开始监听数据变更");

            while (running && !Thread.currentThread().isInterrupted()) {
                try {
                    Message message = connector.getWithoutAck(100);
                    long batchId = message.getId();

                    if (batchId == -1 || message.getEntries().isEmpty()) {
                        Thread.sleep(1000);
                        continue;
                    }
                    processEntries(message.getEntries());
                    connector.ack(batchId);

                } catch (InterruptedException e) {
                    log.info("Canal 线程被中断");
                    Thread.currentThread().interrupt();
                } catch (Exception e) {
                    log.error("处理 Canal 数据异常", e);
                    connector.rollback();  // 异常时回滚
                }
            }
        } catch (Exception e) {
            log.error("Canal 客户端运行异常", e);
        } finally {
            connector.disconnect();
            log.info("Canal 连接已断开");
        }
    }

    private void processEntries(List<CanalEntry.Entry> entries) {

        for (CanalEntry.Entry entry : entries) {
            if (entry.getEntryType() != CanalEntry.EntryType.ROWDATA) {
                continue;
            }
            try {
                CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());

                for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                    // 转换数据
                    GoodsDocument goodsDocument = convert(rowChange.getEventType(), rowData);
                    log.info("处理数据: {}", JSONUtil.toJsonStr(goodsDocument));

                    // 异步处理ES操作
                    processElasticsearchOperation(rowChange.getEventType(), goodsDocument);
                }
            } catch (Exception e) {
                log.error("解析binlog数据异常", e);
            }
        }
    }

    @Async("esTaskExecutor")  // 异步处理ES操作
    public void processElasticsearchOperation(CanalEntry.EventType eventType, GoodsDocument goodsDocument) {
        try {
            switch (eventType) {
                case INSERT, UPDATE -> {
                    goodsDocumentRepository.save(goodsDocument);
                    log.debug("ES {} 成功: {}", eventType, goodsDocument.getId());
                }
                case DELETE -> {
                    goodsDocumentRepository.delete(goodsDocument);
                    log.debug("ES DELETE 成功: {}", goodsDocument.getId());
                }
                default -> log.warn("未知事件类型: {}", eventType);
            }
        } catch (Exception e) {
            log.error("ES 操作失败, ID: {}, 操作: {}", goodsDocument.getId(), eventType, e);
            // 这里可以添加重试逻辑
        }
    }

    // 配置ES操作专用线程池
    @Bean("esTaskExecutor")
    public TaskExecutor esTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5);
        executor.setMaxPoolSize(10);
        executor.setQueueCapacity(100);
        executor.setThreadNamePrefix("es-async-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.setWaitForTasksToCompleteOnShutdown(true);
        executor.setAwaitTerminationSeconds(30);
        executor.initialize();
        return executor;
    }

    private GoodsDocument convert(CanalEntry.EventType eventType, CanalEntry.RowData rowData) {
        GoodsDocument goodsDocument = new GoodsDocument();

        List<CanalEntry.Column> columns = rowData.getAfterColumnsList();
        if (eventType == CanalEntry.EventType.DELETE) {
            columns = rowData.getBeforeColumnsList();
        }

        for (CanalEntry.Column column : columns) {
            String columnName = column.getName();
            String value = column.getValue();

            if (value.isEmpty()) continue;

            try {
                switch (columnName) {
                    case "id" -> goodsDocument.setId(Long.parseLong(value));
                    case "goods_name" -> goodsDocument.setGoodsName(value);
                    case "url" -> goodsDocument.setUrl(value);
                    case "price" -> goodsDocument.setPrice(new BigDecimal(value));
                    case "description" -> goodsDocument.setDescription(value);
                    case "brand" -> goodsDocument.setBrand(value);
                    case "category" -> goodsDocument.setCategory(value);
                    case "stock" -> goodsDocument.setStock(Integer.parseInt(value));
                    case "locked_stock" -> goodsDocument.setLockedStock(Integer.parseInt(value));
                }
            } catch (Exception e) {
                log.warn("字段转换失败, 字段: {}, 值: {}", columnName, value, e);
            }
        }
        return goodsDocument;
    }
}


